package com.zyx.flinkdemo.stream.utils;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;

import java.io.IOException;
import java.io.InputStream;
import java.io.Serializable;
import java.util.Properties;

/**
 * @author zyx
 * @since 2021/5/24 20:56
 * desc:
 */
public class KafkaUtilWithEnv implements Serializable {

    private static final long serialVersionUID = 6898698023547967487L;

    public static FlinkKafkaConsumer<String> getKafkaSource(String topic, String groupId) throws IOException {
        Properties prop = new Properties();
        Properties runProp = new Properties();
        InputStream runPropIuputStream = KafkaUtilWithEnv.class.getClassLoader()
                .getResourceAsStream("run-env.properties");
        runProp.load(runPropIuputStream);
        String runEnv = runProp.getProperty("run.env");
        String kafkaServer;
        switch (runEnv) {
            case "dev":
                kafkaServer = "192.168.31.201:9092,192.168.31.202:9092,10.2.13.203:9092";
                break;
            case "test":
                kafkaServer = "192.168.31.201:9092";
                break;
            default:
                kafkaServer = "";
                break;
        }
        prop.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
                kafkaServer);
        prop.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        return new FlinkKafkaConsumer<String>(topic, new SimpleStringSchema(), prop);
    }
}
